package com.manlitech.manlichat.base.netty.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.manlitech.manlichat.base.exception.ExceptionHandle;
import com.manlitech.manlichat.base.exception.MyException;
import com.manlitech.manlichat.base.netty.message.Message;
import com.manlitech.manlichat.base.netty.message.container.MessageHandler;
import com.manlitech.manlichat.base.netty.message.container.MessageHandlerContainer;
import com.manlitech.manlichat.base.netty.pojo.Invocation;
import com.manlitech.manlichat.base.netty.pool.MyChannelHandlerPool;
import com.manlitech.manlichat.base.netty.util.MessageSendUtil;
import com.manlitech.manlichat.enums.ResultEnums;
import com.manlitech.manlichat.util.JsonUtils;
import com.manlitech.manlichat.util.ResultUtil;
import com.manlitech.manlichat.util.SessionInfoUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * websocket 文本处理 String
 * @author shensg
 */
@Component
@ChannelHandler.Sharable
public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private WebSocketServerHandshaker handshaker;
    private static final String WEB_SOCKET_URL = "ws://127.0.0.1:8098/manlichat/echo";
    private final ExecutorService executor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

    private final String PING = "PING";


    @Autowired
    private MessageHandlerContainer messageHandlerContainer;

    @Autowired
    MessageSendUtil messageSendUtil;

    @Autowired
    ExceptionHandle exceptionHandle;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        logger.info("与客户端建立连接，通道开启！");
        //添加到channelGroup通道组
        MyChannelHandlerPool.channelGroup.add(ctx.channel());
        logger.info("新用户连接: {} , 当前在线用户人数: {}",ctx.channel().id(),MyChannelHandlerPool.channelGroup.size());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("与客户端断开连接，通道关闭！");
        //删除到channelGroup 通道组
        MyChannelHandlerPool.channelGroup.remove(ctx.channel());
        logger.info("用户断开连接:{} , 当前在线用户人数: {}",ctx.channel().id(), MyChannelHandlerPool.channelGroup.size());
    }

    //ping、pong
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //用于触发用户事件，包含触发读空闲、写空闲、读写空闲
        if(evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state()== IdleState.READER_IDLE){
                logger.info(ctx.channel().id().asShortText()+" 读空闲... ");
            }else if (event.state()==IdleState.WRITER_IDLE){
                logger.info(ctx.channel().id().asShortText()+" 写空闲... ");
            }else if (event.state()==IdleState.ALL_IDLE){
                logger.info("channel 读写空闲, 准备关闭当前channel  , 当前UsersChanel的数量: "+MyChannelHandlerPool.channelGroup.size());
                Channel channel = ctx.channel();
                channel.close();
                logger.info("channel 关闭后, UsersChanel的数量: "+MyChannelHandlerPool.channelGroup.size());
            }
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        logger.info(msg.toString());
        // 处理客户端向服务端发起http握手请求的业务
        if(msg instanceof FullHttpRequest){
            handlerHttpRequest(ctx,(FullHttpRequest) msg);
        }
        // 处理webSocket连接业务
        else if(msg instanceof WebSocketFrame)
        {
            handlerWebSocketFrame(ctx,(WebSocketFrame)msg);
        }
    }

    /**
     * 处理客户端向服务端发起的webSocket连接业务
     * @param ctx
     * @param frame
     */
    private void handlerWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
        // 判断是否是关闭webSocket的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }

        // 判断是否是ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        // 判断是否是二进制消息，是则抛出异常
        if (frame instanceof BinaryWebSocketFrame) {
            throw new RuntimeException("【" + this.getClass().getName() + "】不支持二进制消息");
        }

        // 最后使用json格式处理
        String request = ((TextWebSocketFrame) frame).text();
        logger.info("接收到的数据,{}" , request);

        // 判断是否是ping消息 心跳
        if(PING.equals(request)){
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        // 这里是JSON字符数据
        if(!JsonUtils.isJSONString(request)){
            logger.warn(ResultEnums.NOT_JSON_OBJECT.getMsg());
            messageSendUtil.responseTheRequest(ctx.channel(), ResultUtil.error(ResultEnums.NOT_JSON_OBJECT.getCode(),ResultEnums.NOT_JSON_OBJECT.getMsg()) );
            return;
        }

        // JSON内容 业务逻辑处理
        invocationBussinessJson(request,ctx);
    }

    /**
     * JSON内容 业务逻辑处理
     * @param request
     * @param ctx
     */
    void invocationBussinessJson(String request, ChannelHandlerContext ctx){
        // json业务逻辑处理
        JSONObject requestJsonObject=JSON.parseObject(request);
        Invocation invocation = JSON.toJavaObject(requestJsonObject,Invocation.class);
        if(invocation == null){
            logger.error(ResultEnums.ERR_NO_PARAM_TYPE.getMsg());
            messageSendUtil.responseTheRequest(ctx.channel(), ResultUtil.error(ResultEnums.ERR_NO_PARAM_TYPE.getCode(),ResultEnums.ERR_NO_PARAM_TYPE.getMsg()) );
            return;
        }
        // <3.1> 获得 type 对应的 MessageHandler 处理器
        MessageHandler messageHandler = messageHandlerContainer.getMessageHandler(invocation.getType());
        // 获得  MessageHandler 处理器的消息类
        Class<? extends Message> messageClass = MessageHandlerContainer.getMessageClass(messageHandler);
        // <3.2> 解析消息
        Message message = JSON.parseObject(invocation.getMessage(), messageClass);
        // <3.3> 执行逻辑
        executor.submit(new Runnable() {

            @Override
            public void run() {
                // noinspection unchecked
                try {
                    messageHandler.execute(ctx.channel(), message);
                } catch (Exception e) {
                    logger.error("[--错误集中处理 开始--]");
                    messageSendUtil.responseTheRequest(ctx.channel(), exceptionHandle.handle(e));
                    logger.error("[--错误集中处理 结束--]");
                }
            }

        });
    }
    /**
     * 处理客户端向服务端发起的http握手请求的业务
     * @param ctx
     * @param req
     */
    private void handlerHttpRequest(ChannelHandlerContext ctx,FullHttpRequest req){
        // 请求失败或者不是webSocket请求
        if(!req.decoderResult().isSuccess() || !"websocket".equals(req.headers().get("Upgrade"))){
            sendHttpResponse(ctx,req,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(WEB_SOCKET_URL,null,false);
        handshaker = wsFactory.newHandshaker(req);
        if(handshaker == null){
            // 将错误信息返回
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        }else{
            handshaker.handshake(ctx.channel(),req);
        }

    }

    /**
     * 服务端向客户端相应消息
     * @param ctx
     * @param req
     * @param resp
     */
    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse resp){

        if(resp.status().code() != 200){
            ByteBuf buf = Unpooled.copiedBuffer(resp.status().toString(), CharsetUtil.UTF_8);
            // 写入resp
            resp.content().writeBytes(buf);
            buf.release();
        }

        // 发送数据
        ChannelFuture future = ctx.channel().writeAndFlush(resp);
        if(resp.status().code() != 200){
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }
}
